package com.zc.common.mq.product;

import com.alibaba.fastjson.JSON;
import com.zc.common.mq.service.SendMessageDao;
import com.zc.common.mq.utils.MessageUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.UUID;

/**
 * <p>Description : </p>
 *
 * @author : xuhongcao
 * @version : v1.0.0
 * @since : 2/4/2022
 **/
@Slf4j
@Component
public class DefaultRocketProductor implements Productor {

    @Autowired
    SendMessageDao sendMessageDao;
    @Autowired
    private DefaultMQProducer defaultMQProducer;

    @Override
    public <T extends Serializable> void sendMessage(String topic, String tag, T message, int delayLevel) {
        String msg = JSON.toJSONString(message);
        String msgKey = UUID.randomUUID().toString();
        try {
            sendMessageDao.insert(msgKey, topic, tag, delayLevel, msg);
            Message sendMessage = MessageUtil.getMessage(topic, msgKey, tag, msg, delayLevel);
            this.defaultMQProducer.send(sendMessage, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("");
                    sendMessageDao.delete(msgKey);
                }

                @Override
                public void onException(Throwable throwable) {
                    throwable.printStackTrace();
                }
            }, 2000L);
        } catch (Exception var4) {
            log.warn("mq 发送失败");
        }
    }
}
